package com.ule.merchant.order.mon.service.kafka;

import com.ule.merchant.order.mon.service.listener.ApplicationInitBean;
import com.ule.tools.client.kafka.consumer.Consumer;
import com.ule.tools.client.kafka.core.config.KafkaClientsConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @Description: Kafka配置
 */
@Component
@Slf4j
public class KafkaConfig implements ApplicationInitBean {

    public void init() {

        log.info("KafkaConfig init...");

        // zookeeper绕过sasl安全机制，直接向系统获取资源
        System.setProperty("zookeeper.sasl.client", "false");

        // 获取默认路径下的kafka-clients-cfg.xml
        KafkaClientsConfig config = KafkaClientsConfig.DEFAULT_INSTANCE;

        try {
            //接收kafka消息接口
            Consumer consumer = config.newConsumer("order.mon.service.kafka.consumer.KafkaTest");
            // 参数为启动时间,单位为Seconds,如果<0，则一直开下去，直到结束
            consumer.startReceive();
        } catch (Exception e) {
            log.error("error on start kafka order.mon.service.kafka.consumer.KafkaTest", e);
        }

    }

}
